package com.mindata.blockchain.socket.raft.server;

import com.mindata.blockchain.socket.common.Const;
import org.apache.commons.io.FileUtils;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBIterator;
import org.iq80.leveldb.ReadOptions;
import org.iq80.leveldb.impl.Iq80DBFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.Map;

public class RaftStateMachineLevelDB implements RaftStateMachine {
    private static final Logger LOG = LoggerFactory.getLogger(RaftStateMachineLevelDB.class);
    private DB db;
    private String raftDataDir;

    public RaftStateMachineLevelDB(String raftDataDir) {
        this.raftDataDir = raftDataDir;
    }

    @Override
    public void writeSnapshot(String raftDataDir) {
        if (db == null) {
            throw new RuntimeException("level db is null");
        }
        org.iq80.leveldb.Options options = new org.iq80.leveldb.Options();
        options.createIfMissing(true);
        try {
            DB tempdb = Iq80DBFactory.factory.open(new File(raftDataDir), options);
            ReadOptions readOptions = new ReadOptions();
            readOptions.fillCache(false);
            DBIterator iterator = tempdb.iterator(readOptions);
            while (iterator.hasNext()) {
                Map.Entry<byte[], byte[]> item = iterator.next();
                db.put(item.getKey(), item.getValue());
            }
            iterator.close();
            tempdb.close();
        } catch (IOException e) {
            LOG.info(e.getMessage());
        }
    }

    @Override
    public void readSnapshot(String snapshotDir) {
        try {
            // copy snapshot dir to data dir
            if (db != null) {
                db.close();
            }
            String dataDir = raftDataDir + File.separator + "leveldb_data";
            File dataFile = new File(dataDir);
            if (dataFile.exists()) {
                FileUtils.deleteDirectory(dataFile);
            }
            File snapshotFile = new File(snapshotDir);
            if (snapshotFile.exists()) {
                FileUtils.copyDirectory(snapshotFile, dataFile);
            }
            org.iq80.leveldb.Options options = new org.iq80.leveldb.Options();
            options.createIfMissing(true);
            db = Iq80DBFactory.factory.open(new File(raftDataDir), options);
        } catch (Exception ex) {
            LOG.warn("meet exception, msg={}", ex.getMessage());
        }
    }

    @Override
    public void apply(byte[] dataBytes) {
        try {
            ServerMessage.SetRequest request = ServerMessage.SetRequest.parseFrom(dataBytes);
            db.put(request.getKey().getBytes(Const.CHARSET), request.getValue().getBytes(Const.CHARSET));
        } catch (Exception ex) {
            LOG.warn("meet exception, msg={}", ex.getMessage());
        }
    }

    @Override
    public ServerMessage.GetResponse get(ServerMessage.GetRequest request) {
        try {
            ServerMessage.GetResponse.Builder responseBuilder = ServerMessage.GetResponse.newBuilder();
            byte[] keyBytes = request.getKey().getBytes(Const.CHARSET);
            byte[] valueBytes = db.get(keyBytes);
            if (valueBytes != null) {
                String value = new String(valueBytes, Const.CHARSET);
                responseBuilder.setValue(value);
            }
            ServerMessage.GetResponse response = responseBuilder.build();
            return response;
        } catch (Exception e) {
            LOG.warn("read leveldb error, msg={}", e.getMessage());
            return null;
        }
    }
}
